package com.atguigu.hotitems.analysis;

import com.atguigu.hotitems.analysis.pojo.ItemViewCount;
import com.atguigu.hotitems.analysis.pojo.UserBehavior;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.runtime.operators.util.AssignerWithPeriodicWatermarksAdapter;
import org.apache.flink.util.Collector;

import java.io.File;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class HotItems {

    public static void main(String[] args) throws  Exception {

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

       /* String path = HotItems.class.getResource("/").getPath() + File.separator + "UserBehavior.csv";
        // 2. 从csv文件中获取获取数据
        DataStream<String> inputStream = env.readTextFile(path);*/

        // 从kafka
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "192.168.242.130:9092,192.168.242.131:9092,192.168.242.132:9092");
        props.setProperty("group.id", "consumer");
        // 下面是一些次要参数
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("auto.offset.reset", "latest");

        DataStream<String> inputStream = env.addSource(new FlinkKafkaConsumer<>("hotitems",new SimpleStringSchema(), props));

        // 3.转换成pojo，并分配时间戳和watermark
        DataStream<UserBehavior> userBehaviorDataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(new Long(fields[0]), new Long(fields[1]),
                    new Integer(fields[2]), new String(fields[3]), new Long(fields[4]));
        }).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarksAdapter.Strategy<>(
                new BoundedOutOfOrdernessTimestampExtractor<UserBehavior>(Time.of(200, TimeUnit.MILLISECONDS)) {
                    @Override
                    public long extractTimestamp(UserBehavior element) {
                        return element.getTimestamp() * 1000L;
                    }
                }
        ));

        //4. 分组开窗聚合，得到每个窗口各个商品的count值
        // 过滤只保留pv行为
        DataStream<ItemViewCount> windowAggregateStream = userBehaviorDataStream
                .filter(userBehavior -> "pv".equals(userBehavior.getBehavior()))
                // 按照商品ID分组
                .keyBy(UserBehavior::getItemId)
                // 滑动窗口
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .aggregate(new ItemCountAgg(), new WindowItemCountResult());

        // 5. 收集同一窗口的所有商品数据的count数据，排序输出top n
        DataStream<String> resultStream = windowAggregateStream
                // 按照窗口分组
                .keyBy(ItemViewCount::getWindowEnd)
                // 自定义取出前5
                .process(new TopNHotItems(5));

        resultStream.print("result");
        env.execute("hot items analysis");
    }

    // 自定义增量聚合函数
    public static class ItemCountAgg implements AggregateFunction<UserBehavior,Long, Long> {

        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(UserBehavior value, Long accumulator) {
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    // 自定义全窗口函数
    public static class WindowItemCountResult implements WindowFunction<Long, ItemViewCount, Long, TimeWindow> {

        @Override
        public void apply(Long itemId, TimeWindow window, Iterable<Long> input, Collector<ItemViewCount> out) throws Exception {
            Long windowEnd = window.getEnd();
            Long count = input.iterator().next();
            out.collect(new ItemViewCount(itemId, windowEnd, count));
        }
    }

    public static class TopNHotItems extends KeyedProcessFunction<Long, ItemViewCount, String> {

        private Integer topSize;

        // 定义状态列表，保存当前窗口内所有输出的ItemViewCount
        ListState<ItemViewCount> itemViewCountListState;


        public TopNHotItems(int topSize) {
            this.topSize = topSize;
        }

        @Override
        public void processElement(ItemViewCount value, Context ctx, Collector<String> out) throws Exception {
            // 先缓存在List中，并注册定时器
            itemViewCountListState.add(value);
            // 模拟等待，所以这里时间设定的比较短
            ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            itemViewCountListState = getRuntimeContext().getListState(
                    new ListStateDescriptor<ItemViewCount>("item-view-count-list", ItemViewCount.class)
            );
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

            // 定时器触发，当前已收集到的数据，排序输出
            ArrayList<ItemViewCount> itemViewCounts = Lists.newArrayList(itemViewCountListState.get().iterator());

            // 从多到少， 越热门越在前面
            itemViewCounts.sort((a, b) -> -Long.compare(a.getCount(), b.getCount()));
            StringBuffer resultBuffer = new StringBuffer();
            resultBuffer.append("=======").append(System.lineSeparator());
            resultBuffer.append("窗口结束时间").append(new Timestamp(timestamp - 1)).append(System.lineSeparator());
            
            // 遍历输出Top N
            for (int i = 0; i < Math.min(topSize, itemViewCounts.size()); i++) {
                ItemViewCount itemViewCount = itemViewCounts.get(i);
                resultBuffer.append("NO ").append(i+1).append(":");
                resultBuffer.append(" 商品ID = ").append(itemViewCount.getItemId());
                resultBuffer.append(" 热门度 = ").append(itemViewCount.getCount()).append(System.lineSeparator());
            }
            resultBuffer.append("==================").append(System.lineSeparator());

            // 控制输出频率
            Thread.sleep(1000L);

            out.collect(resultBuffer.toString());
        }

    }


}
